package com.atguigu.flink0922.chapter08;

import com.atguigu.flink0922.bean.AdsClickLog;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneOffset;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/9 9:31
 */
public class Flink04_High_Project_BlackList {
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 创建WatermarkStrategy
        WatermarkStrategy<AdsClickLog> wms = WatermarkStrategy
            .<AdsClickLog>forBoundedOutOfOrderness(Duration.ofSeconds(20))
            .withTimestampAssigner(new SerializableTimestampAssigner<AdsClickLog>() {
                @Override
                public long extractTimestamp(AdsClickLog element, long recordTimestamp) {
                    return element.getTimestamp() * 1000L;
                }
            });
        
        final SingleOutputStreamOperator<String> mainStream = env
            .readTextFile("input/AdClickLog.csv")
            .map(line -> {
                String[] datas = line.split(",");
                return new AdsClickLog(Long.valueOf(datas[0]),
                                       Long.valueOf(datas[1]),
                                       datas[2],
                                       datas[3],
                                       Long.valueOf(datas[4]));
            })
            .assignTimestampsAndWatermarks(wms)
            // 统计每个用户每个广告的点击量
            .keyBy(log -> log.getUserId() + "_" + log.getAdsId())
            .process(new KeyedProcessFunction<String, AdsClickLog, String>() {
    
                private ValueState<Boolean> warnState;
                private ValueState<Long> timerState;
                private ReducingState<Long> countState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    countState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Long>(
                        "countState",
                        new ReduceFunction<Long>() {
                            @Override
                            public Long reduce(Long value1, Long value2) throws Exception {
                                return value1 + value2;
                            }
                        },
                        Long.class));
                    
                    timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timerState", Long.class));
    
                    warnState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("warnState", Boolean.class));
                }
                
                @Override
                public void processElement(AdsClickLog input,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    
                    
    
    
                    // 判断是否需要加入黑名单
                    if (countState.get() != null && countState.get() > 99) {
                        
                        if(warnState.value() == null){
                            String blackListMsg = "用户: " + input.getUserId() +
                                "广告: " + input.getAdsId() +
                                "点击量: " + countState.get();
                            ctx.output(new OutputTag<String>("blackList") {}, blackListMsg);
                            warnState.update(true);
                        }
                        
                       
                    } else {
                        countState.add(1L);
                        // 正常流
                        String msg = "用户: " + input.getUserId() +
                            "广告: " + input.getAdsId() +
                            "点击量: " + countState.get();
                        out.collect(msg);
                    }
                    
                    if (timerState.value() == null) {
                        final Long now = ctx.timestamp() / 1000;
                        final LocalDateTime today = LocalDateTime.ofEpochSecond(now, 0, ZoneOffset.ofHours(8));
                        final LocalDateTime tomorrow = LocalDateTime.of(today.toLocalDate().plusDays(1), LocalTime.of(0, 0, 0));
                        final long time = tomorrow.toEpochSecond(ZoneOffset.ofHours(8));
                        
                        ctx.timerService().registerEventTimeTimer(time * 1000);
                        
                        timerState.update(time * 1000);
                    }
                   
                    
                }
    
                @Override
                public void onTimer(long timestamp,
                                    OnTimerContext ctx,
                                    Collector<String> out) throws Exception {
                    countState.clear();
                    timerState.clear();
                    warnState.clear();
                }
            });
    
//        mainStream.print("main");
        mainStream.getSideOutput(new OutputTag<String>("blackList") {}).print("blackList");
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
